package cn.jly.flink.kafka;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * read from kafka topic and write to another kafka topic
 *
 * @author lanyangji
 */
public class FlinkConsumeAndProduceKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        props.put("group.id", "metric-group");
        //key 反序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        //value 反序列化
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("auto.offset.reset", "latest");

        FlinkKafkaConsumer011<String> consumer = new FlinkKafkaConsumer011<>(
                "first",
                new SimpleStringSchema(),
                props
        );
        // consume
        SingleOutputStreamOperator<String> dataStream = env.addSource(consumer).map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                Metric metric = JSON.parseObject(s, Metric.class);
                // 打印一下
                System.out.println(metric);

                // 修改一下，以便于写入另一个topic
                metric.setName("wanghaitao");

                // 再转成JSON
                return JSON.toJSONString(metric);
            }
        });

        // produce
        dataStream.addSink(new FlinkKafkaProducer011<>("second", new SimpleStringSchema(), props));

        env.execute("FlinkConsumeKafka");
    }
}
